package mt.omid.dimomuth.bus;


import static java.util.logging.Level.*;
import static mt.omid.dimomuth.internal.ErrorList.*;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

import javax.security.auth.kerberos.KerberosKey;

import mt.omid.dimomuth.Kernel;
import mt.omid.dimomuth.mgm.MgmHandler;
import mt.omid.dimomuth.mgm.MgmProtocol;
import mt.omid.dimomuth.mgm.MgmServerException;


public class DataBus implements Runnable
{

    // private BlockingQueue<CBMessage> busBuffer;
    //    private Thread	lsnrThrd;
    private static int      port;
    private static int      state;
    private static int      queueSize     = 100;
    private static int      instId;

    public final static int STAT_RUNNING  = 1;
    public final static int STAT_SHUTDOWN = 2;

    private static Logger   logger;
    private static DataBus  instance;
    private static HashMap<Integer, BlockingQueue<CBMessage>> userList;

    static
    {
//	System.out.println( "Start static initializing DatBus..., logger=" + logger );
	logger = Logger.getLogger( DataBus.class.getSimpleName() );
	instance = new DataBus( 100, 0, logger );
	userList = new HashMap<Integer, BlockingQueue<CBMessage>>();
	System.out.println(  "DataBus initialization finished. logger:" + logger );
	logger.fine( "DataBus initialization finished." );
    }


    private DataBus( int busQueueSize, int prt, Logger lgr )
    {
//	System.out.println( "Create DataBus instance..." );
	// busBuffer = new ArrayBlockingQueue<CBMessage>( busBufferSize );
	queueSize = busQueueSize;
	DataBus.port = prt;
//	logger = lgr;
	instId = Kernel.getInstanceId();
//	System.out.println( "Finish DataBus instance creation." );
    }

    public static DataBus getInstance()
    {
	return instance;
    }

    public void setPort( int p )
    {
	port = p;
    }

    public void setQueueSize( int size )
    {
	queueSize = size;
    }

    public static void register( int userId ) throws DataBusException
    {
	if( userList.containsKey( userId ) )
	    throw new DataBusException( "This userid already registered." );
	logger.info( "Registering user id " + userId
		+ " in the user list of data bus and create queue for it in size of " + queueSize );
	userList.put( userId, new ArrayBlockingQueue<CBMessage>( queueSize ) );
    }

    /***
     * Module or other classes who use this facility use the below methods to
     * read or write messages
     */
    public static CBMessage recv( int id, long timeout )
    {
	try
	{
	    return userList.get( id ).poll( timeout, TimeUnit.MILLISECONDS );
	}
	catch( InterruptedException e )
	{
	    return null;
	}
	catch( NullPointerException nex )
	{
	    logger.log( SEVERE, "The reveived message destination doesn't exist:" + id, nex );
	    return null;
	}
    }

    public static CBMessage recvNB( int id )
    {
	try
	{
	    return userList.get( id ).poll();
	}
	catch( NullPointerException nex )
	{
	    logger.log( SEVERE, "The reveived message destination doesn't exist:" + id, nex );
	    return null;
	}
    }

    public static CBMessage recv( int id ) throws InterruptedException
    {
	try
	{
	    return userList.get( id ).take();
	}
	catch( NullPointerException nex )
	{
	    logger.log( SEVERE, "The reveived message destination doesn't exist:" + id, nex );
	    return null;
	}
    }

    public static void send( CBMessage msg ) throws InterruptedException
    {
	// FIXME
	// find the remote side to decide about sending option

	if( msg.rInst == instId ) // internal
	{
	    userList.get( msg.rId ).put( msg );
	}
	else
	{
	    // not implemented yet for network distributed mode
	    throw new RuntimeException( "Send to external instance through DataBus Not Implemented Yet" );
	}
    }

    public static boolean sendNB( CBMessage msg )
    {
	// FIXME
	// find the remote side to decide about sending option

	if( msg.rInst == instId ) // internal
	{
	    return userList.get( msg.rId ).offer( msg );
	}
	else
	{
	    // not implemented yet for network distributed mode
	    throw new RuntimeException( "Send to external instance through DataBus Not Implemented Yet" );
	}

//	return false;
    }

    public static boolean sendNB( CBMessage msg, long TO ) throws InterruptedException
    {
	// FIXME
	// find the remote side to decide about sending option

	if( msg.rInst == instId ) // internal
	{
	    return userList.get( msg.rId ).offer( msg, TO, TimeUnit.MILLISECONDS );
	}
	else
	{
	    // not implemented yet for network distributed mode
	    throw new RuntimeException( "Send to external instance through DataBus Not Implemented Yet" );
	}

//	return false;
    }

    //    public int findRemoteUser( int id )
    //    {
    //	
    //    }

    /***
     * BusProtocol use it to insert received message to this queue
     * 
     * @param msg
     */
    public static void insert( CBMessage msg )
    {
	try
	{
	    userList.get( msg.rId ).put( msg );
	}
	catch( InterruptedException e )
	{
	}
	catch( NullPointerException nex )
	{
	    logger.log( SEVERE, "The reveived message destination doesn't exist:" + msg.rId, nex );
	}
    }

    public void startListening()
    {
	try
	{
	    state = STAT_RUNNING;
	    ServerSocket srvSock = new ServerSocket( port );

	    int i = 0;

	    try
	    {
		while( state == STAT_RUNNING )
		{
		    i++;
		    Socket sock = srvSock.accept();
		    Thread thread = new Thread( new BusProtocol( sock, logger, this ),
			    "DataBusServerThread-" + i );
		    thread.start();
		    logger.info( "New Bus client connected[" + sock.getInetAddress() + ":"
			    + sock.getPort() + "]" + thread.getName() );
		}
	    }
	    catch( Throwable t )
	    {
		logger.log( SEVERE, "Fatal error in data bus listener.", t );
	    }
	}
	catch( IOException e )
	{
	    logger.log( INFO, "Error-" + MGM_SERVER_START_EXCP
		    + ":IOException in starting serve thread.", e );
	    // throw new MgmServerException( "Error-" + MGM_SERVER_START_EXCP
	    // + ":IOException in starting serve thread.", e );
	}
    }

    public void shutdown()
    {
	// lsnrThrd.stop();
	state = STAT_SHUTDOWN;
    }

    public void run()
    {
	try
	{
	    startListening();
	}
	catch( Throwable e )
	{
	    logger.log( WARNING, "Exception in DataBus Server listening.", e );
	}
    }
}
